package com.atguigu.realtime.util

import java.time.LocalDate

import com.atguigu.realtime.bean.StartupLog
import io.searchbox.client.config.HttpClientConfig
import io.searchbox.client.{JestClient, JestClientFactory}
import io.searchbox.core.{Bulk, Index}
import org.apache.spark.rdd.RDD

/**
 * Author atguigu
 * Date 2020/11/13 8:48
 */
object MyEsUtil {
    val factory = new JestClientFactory
    val config = new HttpClientConfig.Builder("http://hadoop162:9200")
        .connTimeout(1000 * 10)
        .readTimeout(1000 * 10)
        .maxTotalConnection(100)
        .multiThreaded(true)
        .build()
    factory.setHttpClientConfig(config)
    
    
    def insertBulk(defaultIndex: String, sources: Iterator[Object]) = {
        val client: JestClient = factory.getObject
        
        val bulkBuilder = new Bulk.Builder()
            .defaultIndex(defaultIndex)
            .defaultType("_doc")
        /*sources
            .map(source => {
                new Index.Builder(source).build()
            })
            .foreach(bulkBuilder.addAction)*/
        
        for (source <- sources) {
            source match {
                case (id: String, s) =>
                    val index: Index = new Index.Builder(s)
                        .id(id)
                        .build()
                    bulkBuilder.addAction(index)
                case s =>
                    val index: Index = new Index.Builder(s)
                        .build()
                    bulkBuilder.addAction(index)
            }
        }
        client.execute(bulkBuilder.build())
        client.shutdownClient()
    }
    
    
    def insertSingle(indexStr: String, source: Object, id: String = null) = {
        val client: JestClient = factory.getObject
        val index: Index = new Index.Builder(source)
            .index(indexStr)
            .`type`("_doc")
            .id(id)
            .build()
        client.execute(index)
        client.shutdownClient()
    }
    
    implicit class RichRDD(rdd: RDD[StartupLog]) {
        def saveToES(index: String) ={
            rdd.foreachPartition((it: Iterator[StartupLog]) => {
                val today: String = LocalDate.now().toString
                MyEsUtil.insertBulk(index, it)
            })
        }
    }
    
}

/*
put /index//typeid
{

}

 */
